package com.源码分析;

import org.junit.Test;
import org.reactivestreams.Subscription;

import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

public class BaseSubscriberDemo {

	@Test
	public void test() {
		SampleSubscriber<Integer> ss = new SampleSubscriber<Integer>();
		Flux<Integer> ints = Flux.range(1, 4);
		ints.subscribe(ss);
	}

}

class SampleSubscriber<T> extends BaseSubscriber<T> {

	@Override
	public void hookOnSubscribe(Subscription subscription) {
		// 操纵request的时候，你必须小心地产生足够的序列要求，否则你的Flux会被卡住。所以，BaseSubscriber的hookOnSubscribe默认是无限的request。如果你覆盖这个钩子，最少要调用一次request
		System.out.println("Subscribed");
		request(1);
	}

	@Override
	public void hookOnNext(T value) {
		System.out.println("SampleSubscriber: " + value);
		request(1);
	}

}